package kafka

import (
	"fmt"
	"github.com/IBM/sarama"
	"go.uber.org/zap"
	"niu-video/pkg/logger"
)

var (
	ProducerId = 1
	ConsumerId = 1
)

type Producer struct {
	Producer   sarama.SyncProducer
	Topic      string //主题
	ProducerID int    //生产者Id
	MessageId  int
}

type Consumer struct {
	Consumer   sarama.Consumer
	Topic      string
	ConsumerId int //消费者Id
}

func (p *Producer) InitProducer(topic string) {
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll          // 发送完数据需要leader和follow都确认
	config.Producer.Partitioner = sarama.NewRandomPartitioner // 新选出一个partition
	config.Producer.Return.Successes = true                   // 成功交付的消息将在success channel返回

	// 连接kafka
	client, err := sarama.NewSyncProducer([]string{"172.31.0.50:9092"}, config)
	if err != nil {
		logger.Error("kafka", zap.String("producer", err.Error()))
		return
	}

	p.Producer = client
	p.Topic = topic
	p.ProducerID = ProducerId
	p.MessageId = 1

	ProducerId++
}

func (c *Consumer) InitConsumer(topic string) error {
	consumer, err := sarama.NewConsumer([]string{"172.31.0.50:9092"}, nil)
	if err != nil {
		return err
	}
	c.Consumer = consumer
	c.Topic = topic
	c.ConsumerId = ConsumerId
	ConsumerId++
	return nil
}

func (p *Producer) SendMessage(txt string) {
	// 构造一个消息
	msg := &sarama.ProducerMessage{}
	msg.Topic = p.Topic
	msg.Value = sarama.StringEncoder(txt)

	// 发送消息
	pid, offset, err := p.Producer.SendMessage(msg)
	//_, _, err := client.SendMessage(msg)
	if err != nil {
		logger.Error("kafka", zap.String("send msg failed, err:", err.Error()))
		return
	}
	logger.Info(fmt.Sprintf("ProducerID:%d pid:%v offset:%v msg:%s",
		p.ProducerID, pid, offset, txt))

	p.MessageId++
}

func (p *Producer) Close() {
	p.Producer.Close()
}

// 指定partition
// offset 可以指定，传-1为获取最新offest
func (c *Consumer) GetMessage(partitionId int32, offset int64) {
	if offset == -1 {
		offset = sarama.OffsetNewest
	}
	pc, err := c.Consumer.ConsumePartition(c.Topic, partitionId, offset)
	if err != nil {
		logger.ErrorString("kafka", "", fmt.Sprintf("failed to start consumer for partition %d,err:%v", partitionId, err))
		return
	}

	// 异步从每个分区消费信息
	go func(sarama.PartitionConsumer) {
		for msg := range pc.Messages() {
			logger.InfoString("kafka", "", fmt.Sprintf("ConsumerId:%d Partition:%d Offset:%d Key:%v Value:%v", c.ConsumerId, msg.Partition, msg.Offset, msg.Key, string(msg.Value)))
		}
	}(pc)
}

// 遍历所有分区
func (c *Consumer) GetMessageToAll(offset int64) {
	partitionList, err := c.Consumer.Partitions(c.Topic) // 根据topic取到所有的分区
	if err != nil {
		logger.Error("kafka", zap.String("kafka", err.Error()))
		return
	}

	for partition := range partitionList { // 遍历所有的分区
		c.GetMessage(int32(partition), -1)
	}
}
